package cn.kent.util;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

import javax.annotation.Resource;

@SuppressWarnings({"all"})
@Component
public class KafkaUtil {
    private static final Logger logger = LoggerFactory.getLogger(KafkaUtil.class);

    @Resource(name = "kafkaTemplate")
    private KafkaTemplate<String, Object> kafkaTemplate;

    public void send(String topic, String message) {
        logger.info("topic=" + topic + ",message=" + message);
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, message);
        future.addCallback(success -> logger.info("KafkaMessageProducer 发送消息成功！"),
                fail -> logger.error("KafkaMessageProducer 发送消息失败！topic:" + topic + "msg:" + message, "KafkaUtil.send"));
    }

}
